package com.example.zyxdemo.consumer;

import lombok.Data;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.acl.AclOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class KafkaTopicMonitor {
    private static final Logger logger = LoggerFactory.getLogger(KafkaTopicMonitor.class);
    private final AdminClient adminClient;
    private final String bootstrapServers;

    public KafkaTopicMonitor(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
        Properties config = new Properties();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        this.adminClient = AdminClient.create(config);
    }

    /**
     * 获取主题的分区信息
     */
    public Map<Integer, PartitionInfo> getTopicPartitions(String topic) {
        Map<Integer, PartitionInfo> partitions = new HashMap<>();
        try {
            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(topic));
            TopicDescription topicDescription = describeTopicsResult.values().get(topic).get();
            for (org.apache.kafka.common.TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                partitions.put(partitionInfo.partition(), new PartitionInfo(
                        partitionInfo.partition(),
                        partitionInfo.leader().id(),
                        partitionInfo.leader().host(),
                        partitionInfo.replicas().stream().map(Node::id).collect(Collectors.toList()),
                        partitionInfo.isr().stream().map(Node::id).collect(Collectors.toList())
                ));
            }
        } catch (InterruptedException | ExecutionException e) {
            logger.error("获取主题分区信息失败", e);
        }
        return partitions;
    }

    /**
     * 获取broker ID到IP的映射
     */
    public Map<Integer, String> getBrokerIps() {
        Map<Integer, String> brokerIps = new HashMap<>();
        try {
            DescribeClusterResult describeClusterResult = adminClient.describeCluster();
            Collection<Node> nodes = describeClusterResult.nodes().get();
            for (Node node : nodes) {
                brokerIps.put(node.id(), node.host());
            }
        } catch (InterruptedException | ExecutionException e) {
            logger.error("获取broker信息失败", e);
        }
        return brokerIps;
    }

    /**
     * 获取消费该主题的消费者组信息
     */
    public Map<String, ConsumerGroupInfo> getConsumerGroups(String topic) {
        Map<String, ConsumerGroupInfo> consumerGroups = new HashMap<>();
        try {
            ListConsumerGroupsResult listConsumerGroupsResult = adminClient.listConsumerGroups();
            Collection<ConsumerGroupListing> groupListings = listConsumerGroupsResult.all().get();

            for (ConsumerGroupListing groupListing : groupListings) {
                try {
                    DescribeConsumerGroupsResult describeConsumerGroupsResult = adminClient.describeConsumerGroups(
                            Collections.singletonList(groupListing.groupId()));
                    ConsumerGroupDescription groupDescription = describeConsumerGroupsResult
                            .all().get().get(groupListing.groupId());

                    // 检查该消费者组是否消费目标主题
                    List<MemberDescription> list = groupDescription.members().stream()
                            .filter(member -> member.assignment()
                                    .topicPartitions().stream()
                                    .anyMatch(tp -> tp.topic().equals(topic)))
                            .collect(Collectors.toList());
                    boolean hasTopic = !CollectionUtils.isEmpty( list);
//                    for (MemberDescription member : groupDescription.members()) {
//                        Collection<TopicPartition> assignments = member.assignment().topicPartitions();
//                        for (TopicPartition assignment : assignments) {
//                            if (assignment.topic().equals(topic)) {
//                                hasTopic = true;
//                                break;
//                            }
//                        }
//                        if (hasTopic) break;
//                    }

                    if (hasTopic) {
                        consumerGroups.put(groupListing.groupId(), new ConsumerGroupInfo(
                                groupListing.groupId(),
                                groupDescription.state().toString(),
                                groupDescription.members().size(),
                                "groupDescription.protocolType()"
                        ));
                    }
                } catch (Exception e) {
                    logger.warn("获取消费者组 {} 信息失败", groupListing.groupId(), e);
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            logger.error("获取消费者组信息失败", e);
        }
        return consumerGroups;
    }

    /**
     * 尝试分析连接到Kafka的生产者IP（注意：这只是一个近似方法）
     */
    public Map<String, List<ConnectionInfo>> analyzeProducerIps() {
        Map<String, List<ConnectionInfo>> producerConnections = new HashMap<>();
        Map<Integer, String> brokerIps = getBrokerIps();

        // 这里是一个简化的示例，实际环境中可能需要更复杂的网络分析
        // 注意：Java没有像Python的psutil那样直接获取网络连接的API
        // 这个方法只是演示如何尝试获取连接信息
        for (Map.Entry<Integer, String> entry : brokerIps.entrySet()) {
            String brokerIp = entry.getValue();
            try (Socket socket = new Socket(brokerIp, 9092)) {
                InetAddress localAddress = socket.getLocalAddress();
                int localPort = socket.getLocalPort();

                // 注意：这只是当前JVM的连接，不是所有生产者连接
                ConnectionInfo connInfo = new ConnectionInfo(
                        localAddress.getHostAddress(),
                        localPort,
                        "Java Producer (approximate)",
                        System.currentTimeMillis()
                );

                producerConnections.computeIfAbsent(brokerIp, k -> new ArrayList<>())
                        .add(connInfo);
            } catch (IOException e) {
                logger.warn("无法连接到broker {}", brokerIp, e);
            }
        }

        // 对于更全面的连接分析，可能需要：
        // 1. 使用JNI调用系统命令（如netstat或ss）
        // 2. 使用像osquery这样的工具
        // 3. 在Kafka broker端配置日志记录连接信息

        return producerConnections;
    }

    /**
     * 尝试分析连接到Kafka的消费者IP（注意：这只是一个近似方法）
     */
    public Map<String, List<ConnectionInfo>> analyzeConsumerIps() {
        // 实现类似于analyzeProducerIps，但针对消费者
        // 注意：Kafka本身不直接提供消费者IP信息
        // 这个方法只是演示如何尝试获取连接信息
        Map<String, List<ConnectionInfo>> consumerConnections = new HashMap<>();
        Map<Integer, String> brokerIps = getBrokerIps();

        for (Map.Entry<Integer, String> entry : brokerIps.entrySet()) {
            String brokerIp = entry.getValue();
            try (Socket socket = new Socket(brokerIp, 9092)) {
                InetAddress localAddress = socket.getLocalAddress();
                int localPort = socket.getLocalPort();

                // 注意：这只是当前JVM的连接，不是所有消费者连接
                ConnectionInfo connInfo = new ConnectionInfo(
                        localAddress.getHostAddress(),
                        localPort,
                        "Java Consumer (approximate)",
                        System.currentTimeMillis()
                );

                consumerConnections.computeIfAbsent(brokerIp, k -> new ArrayList<>())
                        .add(connInfo);
            } catch (IOException e) {
                logger.warn("无法连接到broker {}", brokerIp, e);
            }
        }

        return consumerConnections;
    }

    /**
     * 获取增强的消费者组信息，尝试包含IP地址
     */
    public Map<String, EnhancedConsumerGroupInfo> getEnhancedConsumerGroups(String topic) {
        Map<String, EnhancedConsumerGroupInfo> groups = new HashMap<>();

        try {
            // 获取基本的消费者组信息
            Map<String, ConsumerGroupInfo> basicGroups = getConsumerGroups(topic);

            for (Map.Entry<String, ConsumerGroupInfo> entry : basicGroups.entrySet()) {
                String groupId = entry.getKey();
                ConsumerGroupInfo basicInfo = entry.getValue();

                // 获取详细的成员信息
                DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
                        Collections.singletonList(groupId));
                ConsumerGroupDescription groupDesc = describeResult.all().get().get(groupId);

                List<EnhancedConsumerMember> members = new ArrayList<>();
                for (MemberDescription member : groupDesc.members()) {
                    // 尝试从客户端元数据中提取IP
                    String clientId = member.clientId();
                    String host = member.host(); // 注意：这可能是主机名而非IP
                    String memberId = member.consumerId();

                    // 尝试解析主机名为IP
                    String ipAddress = host;
                    try {
                        InetAddress inetAddress = InetAddress.getByName(host);
                        ipAddress = inetAddress.getHostAddress();
                    } catch (UnknownHostException e) {
                        logger.warn("无法解析主机名: {}", host);
                    }
                    if(member.assignment().topicPartitions().stream().anyMatch(tp -> tp.topic().equals(topic))){
                        members.add(new EnhancedConsumerMember(
                                memberId,
                                clientId,
                                ipAddress,
                                member.assignment().topicPartitions()
                        ));
                    }

                }

                groups.put(groupId, new EnhancedConsumerGroupInfo(
                        groupId,
                        basicInfo.getState(),
                        basicInfo.getMemberCount(),
                        basicInfo.getProtocolType(),
                        members
                ));
            }
        } catch (Exception e) {
            logger.error("获取增强的消费者组信息失败", e);
        }

        return groups;
    }

    /**
     * 关闭Kafka连接
     */
    public void close() {
        if (adminClient != null) {
            adminClient.close(Duration.ofSeconds(10));
        }
    }

    @Data
    // 增强的消费者组成员信息
    public static class EnhancedConsumerMember {
        private final String memberId;
        private final String clientId;
        private final String ipAddress;
        private final Collection<TopicPartition> assignments;

        // 构造函数和getters
    }
    @Data
    // 增强的消费者组信息
    public static class EnhancedConsumerGroupInfo {
        private final String groupId;
        private final String state;
        private final int memberCount;
        private final String protocolType;
        private final List<EnhancedConsumerMember> members;

        // 构造函数和getters
    }
    // 内部类：分区信息
    public static class PartitionInfo {
        private final int partitionId;
        private final int leaderId;
        private final String leaderHost;
        private final List<Integer> replicas;
        private final List<Integer> isr;

        public PartitionInfo(int partitionId, int leaderId, String leaderHost, List<Integer> replicas, List<Integer> isr) {
            this.partitionId = partitionId;
            this.leaderId = leaderId;
            this.leaderHost = leaderHost;
            this.replicas = replicas;
            this.isr = isr;
        }

        // getters
        public int getPartitionId() { return partitionId; }
        public int getLeaderId() { return leaderId; }
        public String getLeaderHost() { return leaderHost; }
        public List<Integer> getReplicas() { return replicas; }
        public List<Integer> getIsr() { return isr; }
    }

    // 内部类：消费者组信息
    public static class ConsumerGroupInfo {
        private final String groupId;
        private final String state;
        private final int memberCount;
        private final String protocolType;

        public ConsumerGroupInfo(String groupId, String state, int memberCount, String protocolType) {
            this.groupId = groupId;
            this.state = state;
            this.memberCount = memberCount;
            this.protocolType = protocolType;
        }

        // getters
        public String getGroupId() { return groupId; }
        public String getState() { return state; }
        public int getMemberCount() { return memberCount; }
        public String getProtocolType() { return protocolType; }
    }

    // 内部类：连接信息
    public static class ConnectionInfo {
        private final String ipAddress;
        private final int port;
        private final String processName;
        private final long connectionTime;

        public ConnectionInfo(String ipAddress, int port, String processName, long connectionTime) {
            this.ipAddress = ipAddress;
            this.port = port;
            this.processName = processName;
            this.connectionTime = connectionTime;
        }

        // getters
        public String getIpAddress() { return ipAddress; }
        public int getPort() { return port; }
        public String getProcessName() { return processName; }
        public long getConnectionTime() { return connectionTime; }
    }

    public static void main(String[] args) {
        // 配置Kafka服务器地址
        String bootstrapServers = "10.160.1.15:9092";
        String topic = "dc_lower_business"; // 替换为实际的topic名称

        KafkaTopicMonitor monitor = new KafkaTopicMonitor(bootstrapServers);

        System.out.println("Kafka主题监控: " + topic);
        System.out.println("=".repeat(50));

        // 获取分区信息
        Map<Integer, PartitionInfo> partitions = monitor.getTopicPartitions(topic);
        Map<Integer, String> brokerIps = monitor.getBrokerIps();
        Map<String, EnhancedConsumerGroupInfo> map = monitor.getEnhancedConsumerGroups(topic);
        System.out.println("\n分区信息:");
        for (Map.Entry<Integer, PartitionInfo> entry : partitions.entrySet()) {
            PartitionInfo info = entry.getValue();
            System.out.printf("分区 %d: 领导者 = %d (%s)%n",
                    info.getPartitionId(), info.getLeaderId(), info.getLeaderHost());
            System.out.printf("  副本: %s%n", info.getReplicas());
            System.out.printf("  同步副本: %s%n", info.getIsr());
        }

        // 获取消费者组信息
        Map<String, ConsumerGroupInfo> consumerGroups = monitor.getConsumerGroups(topic);
        System.out.println("\n消费者组信息:");
        for (Map.Entry<String, ConsumerGroupInfo> entry : consumerGroups.entrySet()) {
            ConsumerGroupInfo info = entry.getValue();
            System.out.printf("消费者组: %s%n", info.getGroupId());
            System.out.printf("  状态: %s%n", info.getState());
            System.out.printf("  成员数: %d%n", info.getMemberCount());
            System.out.printf("  协议类型: %s%n", info.getProtocolType());
        }

        // 获取消费者IP信息
        Map<String, List<ConnectionInfo>> consumerIps = monitor.analyzeConsumerIps();
        System.out.println("\n消费者IP信息:");
        for (Map.Entry<String, List<ConnectionInfo>> entry : consumerIps.entrySet()) {
            String brokerIp = entry.getKey();
            System.out.printf("Broker: %s%n", brokerIp);
            for (ConnectionInfo consumer : entry.getValue()) {
                System.out.printf("  IP: %s:%d, 进程: %s%n",
                        consumer.getIpAddress(),
                        consumer.getPort(),
                        consumer.getProcessName());
            }
        }

        // 获取生产者IP信息
        Map<String, List<ConnectionInfo>> producerIps = monitor.analyzeProducerIps();
        System.out.println("\n生产者IP信息:");
        for (Map.Entry<String, List<ConnectionInfo>> entry : producerIps.entrySet()) {
            String brokerIp = entry.getKey();
            System.out.printf("Broker: %s%n", brokerIp);
            for (ConnectionInfo producer : entry.getValue()) {
                System.out.printf("  IP: %s:%d, 进程: %s%n",
                        producer.getIpAddress(),
                        producer.getPort(),
                        producer.getProcessName());
            }
        }

        monitor.close();
    }
}